package cn.spark.sparkStudy.core;

import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import groovy.lang.Tuple;
import scala.Tuple2;

/**
 * @Author He Bao Jing Z003NUJF
 * @Date Aug 14, 2017
 *
 */

public class Transformation {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		SparkConf conf = new SparkConf().setAppName("WordCount App").setMaster("local");
		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> linesFile = sc.textFile("D://code//data//name.txt");
		List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5);
		JavaRDD<Integer> numberRDD = sc.parallelize(numberList, 1);

		List<String> strList = Arrays.asList("hello u", "world www bob", "bob sucy");
		JavaRDD<String> strRDD = sc.parallelize(strList, 1);
		
		List<Tuple2<String, Integer>> scoreList = Arrays.asList(
				new Tuple2<String, Integer>("class1", 20),
				new Tuple2<String, Integer>("class2", 10),
				new Tuple2<String, Integer>("class2", 60),
				new Tuple2<String, Integer>("class1", 90));
		JavaPairRDD<String, Integer> scoreRDD = sc.parallelizePairs(scoreList);
		
//		filter(numberRDD);
//		flatMap(linesFile);
		reduceByKey(scoreRDD);
		
		sc.close();
	}

	//过滤，得到偶数
	public static void filter(JavaRDD<Integer> numberRDD) {
		JavaRDD<Integer> evenNUmber = numberRDD.filter(new Function<Integer, Boolean>() {

			public Boolean call(Integer value) throws Exception {
				// TODO Auto-generated method stub
				return value % 2 == 0;
			}
		});
		
		evenNUmber.foreach(new VoidFunction<Integer>() {
			public void call(Integer value) throws Exception {
				// TODO Auto-generated method stub
				System.out.print("*" + value);
			}
		});
	}

	//将文本拆分成多个单词
	public static void flatMap(JavaRDD<String> linesFile) {
		JavaRDD<String> words = linesFile.flatMap(new FlatMapFunction<String, String>() {

			public Iterable<String> call(String line) throws Exception {
				// TODO Auto-generated method stub
				return Arrays.asList(line.split(" "));
			}
		});
		
		words.foreach(new VoidFunction<String>() {
			public void call(String word) throws Exception {
				// TODO Auto-generated method stub
				System.out.println(" " + word);
			}
		});
	}
	
	//班级总分数
	public static void reduceByKey(JavaPairRDD<String, Integer> scoreRDD) {
		
		JavaPairRDD<String, Integer> pair = scoreRDD.reduceByKey(
				new Function2<Integer, Integer, Integer>() {
			
			public Integer call(Integer v0, Integer v1) throws Exception {
				// TODO Auto-generated method stub
				return v1 + v0;
			}
		});
		
		pair.foreach(new VoidFunction<Tuple2<String,Integer>>() {

			@Override
			public void call(Tuple2<String, Integer> pair) throws Exception {
				System.out.println(pair._1 + ":" + pair._2);
				
			}
			
		});
		
	}
	
	
}
